package com.zhuziym.utils;

import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import org.apache.commons.lang.builder.ToStringBuilder;
import org.pentaho.di.core.KettleEnvironment;
import org.pentaho.di.core.RowSet;
import org.pentaho.di.core.database.DatabaseMeta;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.logging.LogChannelInterface;
import org.pentaho.di.core.logging.LogLevel;
import org.pentaho.di.i18n.BaseMessages;
import org.pentaho.di.job.Job;
import org.pentaho.di.job.JobEntryResult;
import org.pentaho.di.job.JobMeta;
import org.pentaho.di.repository.RepositoryDirectoryInterface;
import org.pentaho.di.repository.kdr.KettleDatabaseRepository;
import org.pentaho.di.repository.kdr.KettleDatabaseRepositoryMeta;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransMeta;
import org.pentaho.di.trans.step.StepInterface;
import org.pentaho.di.trans.step.StepMeta;
import org.pentaho.di.trans.step.StepMetaDataCombi;
import org.pentaho.di.trans.step.StepStatus;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @Title: KettleUtils.java
 * @Package com.zhuziym.utils
 * @Description: TODO(kettle工具)
 * @author 作者 grq
 * @version 创建时间：2018年11月3日 下午1:49:41
 *
 */
public class KettleUtils {

	private static Class<?> PKG = Trans.class; // for i18n purposes, needed by

	private static Logger log = LoggerFactory.getLogger(KettleUtils.class);

	private KettleUtils() throws Exception {
		throw new Exception("工具类不允许创建");
	}

	/**
	 * kettle 资源库
	 */
	public static KettleDatabaseRepository kettleDbRepository;

	static {
		if (kettleDbRepository == null) {
			kettleDbRepository = getKettleDatabaseRepository();
			log.info("实例化kettle资源库");
		}
	}

	/**
	 * 设置kettle资源库连接信息
	 * 
	 * @return
	 */
	public static DatabaseMeta setDatabaseMeta() {
		return new DatabaseMeta(PropUtils.getConfigValByKey("name"), PropUtils.getConfigValByKey("type"), PropUtils.getConfigValByKey("access"), PropUtils.getConfigValByKey("host"), PropUtils.getConfigValByKey("db"), PropUtils.getConfigValByKey("port"), PropUtils.getConfigValByKey("user"), PropUtils.getConfigValByKey("pass"));
	}

	/**
	 * kettle 资源库 获取
	 * 
	 * @return
	 */
	private static KettleDatabaseRepository getKettleDatabaseRepository() {
		try {

			KettleEnvironment.init();
			// KettleEnvironment.shutdown();
			// 创建DB资源库
			KettleDatabaseRepository repository = new KettleDatabaseRepository();
			DatabaseMeta databaseMeta = KettleUtils.setDatabaseMeta();
			// 选择资源库 此处参我与本地不一致,但还是正确执行
			KettleDatabaseRepositoryMeta kettleDatabaseRepositoryMeta = new KettleDatabaseRepositoryMeta("Kettle", "Kettle", "Transformation description", databaseMeta);
			repository.init(kettleDatabaseRepositoryMeta);
			// 连接资源库 DatabaseMeta中已经设置了，不明白此处为何还要再次设置
			repository.connect("admin", "admin");
			return repository;
		} catch (KettleException e) {
			e.printStackTrace();
			log.error("kettle资源库连接失败");
		}
		return null;
	}

	/**
	 * 执行 JOB
	 * 
	 * @param repository
	 *            kettle数据库资源库
	 * @param job_path
	 *            JOB 目录
	 * @param job_name
	 *            JOB 名称
	 * @param variableMap
	 *            变量Map
	 * @param parameterValueMap
	 *            参数Map
	 * @throws KettleException
	 */
	public static boolean execJob(KettleDatabaseRepository repository, String job_path, String job_name, Map<String, Object> variableMap, Map<String, Object> parameterValueMap) throws KettleException {
		System.setProperty("KETTLE_DEFAULT_SERVLET_ENCODING", StandardCharsets.UTF_8.name());
		RepositoryDirectoryInterface dir = repository.findDirectory(job_path);
		JobMeta jobMeta = repository.loadJob(repository.getJobId(job_name, dir), null);
		Job job = new Job(repository, jobMeta);

		job.setLogLevel(LogLevel.BASIC);//MINIMAL 有时候获取不到日志信息
		// 设置参数
		if (parameterValueMap != null) {
			Set<Entry<String, Object>> entrySet = parameterValueMap.entrySet();
			for (Entry<String, Object> entry : entrySet) {
				job.setParameterValue(entry.getKey(), entry.getValue().toString());
			}
		}
		// 设置变量 java代码中变量会覆盖kettle里面设置的变量
		if (variableMap != null) {
			Set<Entry<String, Object>> entrySet = variableMap.entrySet();
			for (Entry<String, Object> entry : entrySet) {
				job.setVariable(entry.getKey(), entry.getValue().toString());
			}
		}
		// 启动执行指定的job
		job.start();
		job.waitUntilFinished();// 等待job执行完
		job.setFinished(true);
		if (job.isFinished()) {
			List<JobEntryResult> jobEntryResults = job.getJobEntryResults();
			for (JobEntryResult jobEntryResult : jobEntryResults) {
				System.out.println("执行步骤=" + jobEntryResult.getJobEntryName() + ",执行状态=" + jobEntryResult.getResult().getResult());
			}

		}
		// 执行 打印的日志 等全在这个xml中
		System.out.println("------------------------------------------------------------");
		job.getResult().getBasicXml();
		System.out.println(job.getResult().getLogText());
		return job.getResult().getResult();
	}

	/**
	 * 执行 转换
	 * 
	 * @param repository
	 *            数据库资源库
	 * @param path
	 *            转换路径
	 * @param tranName
	 *            转换名称
	 * @param variableMap
	 *            参数Map
	 * @param parameterValueMap
	 *            参数Map
	 * @throws KettleException
	 */
	public static boolean execTran(KettleDatabaseRepository repository, String path, String tranName, Map<String, Object> variableMap, Map<String, Object> parameterValueMap) throws KettleException {
		System.setProperty("KETTLE_DEFAULT_SERVLET_ENCODING", StandardCharsets.UTF_8.name());
		// 根据指定的字符串路径 找到目录
		RepositoryDirectoryInterface dir = repository.findDirectory(path);
		// 在 指定目录下找转换
		TransMeta transMeta = repository.loadTransformation(repository.getTransformationID(tranName, dir), null);
		// 设置参数
		if (parameterValueMap != null) {
			Set<Entry<String, Object>> entrySet = parameterValueMap.entrySet();
			for (Entry<String, Object> entry : entrySet) {
				transMeta.setParameterValue(entry.getKey(), entry.getValue().toString());
			}
		}
		// 设置变量 java代码中变量会覆盖kettle里面设置的变量
		if (variableMap != null) {
			Set<Entry<String, Object>> entrySet = variableMap.entrySet();
			for (Entry<String, Object> entry : entrySet) {
				transMeta.setVariable(entry.getKey(), entry.getValue().toString());
			}
		}
		Trans trans = new Trans(transMeta);
		trans.setLogLevel(LogLevel.BASIC);
		trans.execute(null);// 执行trans
		trans.waitUntilFinished();// 等待直到数据结束

		LogChannelInterface log = trans.getLogChannel();
		log.logBasic(BaseMessages.getString(PKG, "Trans.Log.DispacthingStartedForTransformation", trans.getFilename()));

		// 几个步骤
		List<StepMeta> hopsteps = transMeta.getTransHopSteps(false);
		for (int i = 0; i < hopsteps.size(); i++) {
			StepMeta thisStep = hopsteps.get(i);
			thisStep.getName();// 步骤名称
		}
		// 每个步骤名称
		// StepMetaDataCombi
		List<StepMetaDataCombi> steps = trans.getSteps();
		for (StepMetaDataCombi stepMetaDataCombi : steps) {
			System.out.println(ToStringBuilder.reflectionToString(stepMetaDataCombi));
		}
		// kettle 界面执行面板的 步骤度量
		System.out.println("kettle 界面执行面板的 步骤度量");
		for (int i = 0; i < trans.nrSteps(); i++) {
			StepInterface baseStep = trans.getRunThread(i);
			StepStatus stepStatus = new StepStatus(baseStep);
			// System.out.println(ToStringBuilder.reflectionToString(stepStatus));
			System.out.println("stepname=" + stepStatus.getStepname() + ",errors=" + stepStatus.getErrors() + ",statusDescription=" + stepStatus.getStatusDescription());
		}
		// 转换里面箭头指向
		List<RowSet> rowsets = trans.getRowsets();
		// 转换里面箭头指向 end
		if (trans.getErrors() > 0) {
			System.out.println("transformation error");
			return false;
		}

		System.out.println("transformation successfully");
		return true;
	}
}
